Geospatial support for Snowflake #516
Conversation
Amogh-Bharadwaj
commented
Oct 15, 2023
Amogh-Bharadwaj
commented
- Supports GEOGRAPHY and GEOMETRY of PostGIS from PostgreSQL to Snowflake - both CDC and QRep
- Supports Postgres' POINT data type to Snowflake
- Adds unique timestamp in prefix for S3 test
flow/model/conversion_avro.go
Outdated
| @@ -75,6 +76,7 @@ func GetAvroSchemaDefinition( | |||
| nullableFields := map[string]bool{} | |||
|
|
|||
| for _, qField := range qRecordSchema.Fields { | |||
| log.Infof("qField name: %s, qField type: %s", qField.Name, qField.Type) | |||
| @@ -0,0 +1,6 @@ | |||
| package model | |||
|
|
|||
| type ColumnInformation struct { | |||
There was a problem hiding this comment.
add comments on what column map example would look like
flow/connectors/utils/postgres.go
Outdated
|
|
||
| var qValueKind qvalue.QValueKind | ||
| switch typeName { | ||
| case "geometry": |
There was a problem hiding this comment.
Point in PostGIS comes under GEOMETRY. Example : GEOMETRY('POINT(1 2)'), like GEOMETRY('LINESTRING(1 2, 3 4)')
The point data type I'm adding support for in this PR is Postgres' inbuilt point type which is not custom
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
| @@ -395,7 +413,7 @@ func (qe *QRepQueryExecutor) ExecuteAndProcessQueryStream( | |||
| return totalRecordsFetched, nil | |||
| } | |||
|
|
|||
| func mapRowToQRecord(row pgx.Rows, fds []pgconn.FieldDescription) (*model.QRecord, error) { | |||
| func mapRowToQRecord(row pgx.Rows, fds []pgconn.FieldDescription, customTypeMap map[uint32]string) (*model.QRecord, error) { | |||
There was a problem hiding this comment.
🚫 [golangci] reported by reviewdog 🐶
line is 124 characters (lll)
| case "NUMBER": | ||
| transformations = append(transformations, fmt.Sprintf("$1:\"%s\" AS \"%s\"", colName, colName)) | ||
| default: | ||
| transformations = append(transformations, fmt.Sprintf("($1:\"%s\")::%s AS \"%s\"", colName, colType, colName)) |
There was a problem hiding this comment.
🚫 [golangci] reported by reviewdog 🐶
line is 122 characters (lll)
| return qvalue.QValueKindInvalid | ||
|
|
There was a problem hiding this comment.
🚫 [golangci] reported by reviewdog 🐶
unnecessary trailing newline (whitespace)
| Kind: customTypeToQKind(typeName), | ||
| Value: values[i], | ||
| } | ||
| record.Set(i, *&customTypeVal) |
There was a problem hiding this comment.
🚫 [golangci] reported by reviewdog 🐶
SA4001: *&x will be simplified to x. It will not copy x. (staticcheck)
Amogh-Bharadwaj
force-pushed
the
postgis-support
branch
2 times, most recently
from
92f36f6 to
9618cb8
Compare
flow/e2e/test_utils.go
Outdated
| '{"key": "value"}', 15 | ||
| '{"key": "value"}', 15, | ||
| 'POINT(1 2)','POINT(40.7128 -74.0060)', | ||
| 'LINESTRING(0 0, 1 1, 2 2)','LINESTRING(-74.0060 40.7128, -73.9352 40.7306, -73.9123 40.7831)', |
There was a problem hiding this comment.
🚫 [golangci] reported by reviewdog 🐶
line is 123 characters (lll)
flow/e2e/test_utils.go
Outdated
| '{"key": "value"}', 15, | ||
| 'POINT(1 2)','POINT(40.7128 -74.0060)', | ||
| 'LINESTRING(0 0, 1 1, 2 2)','LINESTRING(-74.0060 40.7128, -73.9352 40.7306, -73.9123 40.7831)', | ||
| 'POLYGON((0 0, 0 1, 1 1, 1 0, 0 0))','POLYGON((-74.0060 40.7128, -73.9352 40.7306, -73.9123 40.7831, -74.0060 40.7128))' |
There was a problem hiding this comment.
🚫 [golangci] reported by reviewdog 🐶
line is 148 characters (lll)
Amogh-Bharadwaj
force-pushed
the
postgis-support
branch
from
9e7833a to
d64aaba
Compare
Similar to #516 Leverages https://cloud.google.com/bigquery/docs/geospatial-data to implement syncing of Postgres' POSTGIS types to BigQuery's GEOGRAPHY data type - for both QRep, Initial Load and CDC. In `qrep_avro_sync`, we now have a function where we can perform transformations of the data on staging table before copying to destination - a feature which was needed here, and makes it easier to support datatypes in QRep for BQ Tests added for QRep and CDC
